/**
 * Copyright 2015 ABSir's Studio
 * 
 * All right reserved
 *
 * Create on 2015年4月13日 下午3:16:24
 */
package com.absir.appserv.master.service;

import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.List;

import org.hibernate.Session;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.absir.appserv.data.value.DataQuery;
import com.absir.appserv.master.bean.JSlave;
import com.absir.appserv.master.bean.JSlaveServer;
import com.absir.appserv.master.bean.JSlaveServerChannel;
import com.absir.appserv.master.bean.JSlaveSynch;
import com.absir.appserv.master.bean.base.ISynchOpition;
import com.absir.appserv.master.bean.base.JbBeanTargets;
import com.absir.appserv.system.adapter.SocketAdapter;
import com.absir.appserv.system.bean.JEmbedSS;
import com.absir.appserv.system.converter.BodyMsgPack.CallbackMessagePack;
import com.absir.appserv.system.dao.BeanDao;
import com.absir.appserv.system.dao.utils.QueryDaoUtils;
import com.absir.appserv.system.helper.HelperDatabind;
import com.absir.appserv.system.service.MasterService;
import com.absir.async.value.Async;
import com.absir.bean.basis.Base;
import com.absir.bean.core.BeanFactoryUtils;
import com.absir.bean.inject.value.Bean;
import com.absir.bean.inject.value.Inject;
import com.absir.context.core.ContextUtils;
import com.absir.context.schedule.value.Schedule;
import com.absir.core.base.Environment;
import com.absir.core.kernel.KernelString;
import com.absir.core.util.UtilAtom;
import com.absir.orm.hibernate.boost.IEntityMerge;
import com.absir.orm.hibernate.boost.L2CacheCollectionService;
import com.absir.orm.transaction.value.Transaction;
import com.absir.server.socket.SocketChannelContext;

/**
 * @author absir
 *
 */
@SuppressWarnings("unchecked")
@Base
@Bean
public abstract class MasterSlaveService implements IEntityMerge<JSlaveServer> {

	/** LOGGER */
	protected static final Logger LOGGER = LoggerFactory.getLogger(MasterSlaveService.class);

	/** ME */
	public static final MasterSlaveService ME = BeanFactoryUtils.get(MasterSlaveService.class);

	/**
	 * @param eids
	 * @return
	 */
	@DataQuery("SELECT o.host.id FROM JSlaveServer o WHERE o.id IN (:p0)")
	public abstract String[] getTargetIds(long[] eids);

	/**
	 * @param entityClass
	 */
	public static <T extends JbBeanTargets> void addEntityMerge(Class<T> entityClass) {
		L2CacheCollectionService.ME.addEntityMerges(entityClass, new IEntityMerge<T>() {

			@Override
			public void merge(String entityName, T entity, com.absir.orm.hibernate.boost.IEntityMerge.MergeType mergeType,
					Object mergeEvent) {
				if (entity.getId() != null) {
					MasterSlaveService.ME.addSlaveSynch(entity.getTargets(), entityName + "@" + entity.getId(), "api/slave/option/"
							+ entityName + "/" + (mergeType == MergeType.DELETE ? 2 : 0), entity);
				}
			}

		});
	}

	/**
	 * 
	 */
	@Transaction
	@Inject
	public void initService() {
		QueryDaoUtils.createQueryArray(BeanDao.getSession(), "UPDATE JSlave o SET o.connecting = FALSE").executeUpdate();
	}

	/**
	 * @param id
	 * @param params
	 * @param socketChannel
	 */
	@Transaction
	public void registerSlave(String id, String[] params, SocketChannel socketChannel) {
		Session session = BeanDao.getSession();
		JSlave slave = BeanDao.get(session, JSlave.class, id);
		if (slave == null) {
			slave = new JSlave();
			slave.setId(id);
			id = null;
		}

		if (params.length > 1) {
			slave.setGroupId(params[1]);
		}

		if (params.length > 2) {
			slave.setVersion(params[2]);
		}

		slave.setIp(socketChannel.socket().getInetAddress().getHostAddress());
		slave.setConnecting(true);
		slave.setLastConnectTime(ContextUtils.getContextTime());
		slave.setSlaveKey(MasterService.ME.buildSlaveKey(id, socketChannel));
		if (id == null) {
			session.persist(slave);
			checkSlaveSynch(slave);

		} else {
			session.merge(slave);
		}
	}

	/**
	 * @param id
	 */
	@Transaction
	public void unRegisterSlave(String id) {
		if (Environment.isStarted()) {
			Session session = BeanDao.getSession();
			JSlave slave = BeanDao.get(session, JSlave.class, id);
			if (slave != null) {
				slave.setConnecting(false);
				slave.setLastConnectTime(ContextUtils.getContextTime());
				session.merge(slave);
				session.flush();
				if (MasterService.ME.getSlaveKey(id) != null) {
					session.getTransaction().rollback();
				}

				session.evict(slave);
			}
		}
	}

	/**
	 * 添加同步
	 * 
	 * @param eids
	 * @param mid
	 * @param uri
	 * @param postData
	 */
	@Transaction
	public void addSlaveSynch(long[] eids, String mid, String uri, Object postData) {
		if (eids == null || eids.length == 0) {
			addSlaveSynch("*", mid, uri, postData);

		} else {
			for (String eid : ME.getTargetIds(eids)) {
				addSlaveSynch(eid, mid, uri, postData);
			}
		}
	}

	/**
	 * 添加同步
	 * 
	 * @param eid
	 * @param mid
	 * @param uri
	 * @param postData
	 */
	@Transaction
	public boolean addSlaveSynch(String eid, String mid, String uri, Object postData) {
		return addSlaveSynch(eid, mid, uri, postData, postData == null || postData instanceof ISynchOpition ? false : true);
	}

	@Transaction
	public boolean addSlaveSynch(String eid, String mid, String uri, Object postData, boolean autoSynch) {
		if (KernelString.isEmpty(eid)) {
			eid = "*";
		}

		JEmbedSS embedSS = new JEmbedSS();
		embedSS.setEid(eid);
		embedSS.setMid(mid);
		JSlaveSynch slaveSynch = new JSlaveSynch();
		slaveSynch.setId(embedSS);
		slaveSynch.setUri(uri);
		slaveSynch.setUpdateTime(System.currentTimeMillis());
		slaveSynch.setSynched("*".equals(eid));
		slaveSynch.setAutoSynched(autoSynch);
		try {
			slaveSynch.setPostData(postData == null ? null : postData.getClass() == byte[].class ? (byte[]) postData
					: HelperDatabind.writeAsBytes(postData));
			BeanDao.getSession().merge(slaveSynch);
			if (slaveSynch.isSynched()) {
				ME.addSlaveSynch(slaveSynch);

			} else {
				ME.checkSyncs();
			}

			return true;

		} catch (Exception e) {
			LOGGER.error("addSlaveSynch failed " + uri + " => " + postData, e);
		}

		return false;
	}

	/**
	 * 添加通用同步
	 * 
	 * @param slaveSynch
	 */
	@Async
	@Transaction
	protected void addSlaveSynch(JSlaveSynch slaveSynch) {
		Iterator<JSlave> iterator = QueryDaoUtils.createQueryArray(BeanDao.getSession(), "SELECT o FROM JSlave o").iterate();
		while (iterator.hasNext()) {
			JSlave slave = iterator.next();
			addSlaveSynch(slave.getId(), slaveSynch.getId().getMid(), slaveSynch.getUri(), slaveSynch.getPostData());
		}
	}

	/**
	 * 检查通用同步
	 * 
	 * @param slave
	 */
	@Async
	@Transaction
	public void checkSlaveSynch(JSlave slave) {
		Iterator<JSlaveSynch> iterator = QueryDaoUtils.createQueryArray(BeanDao.getSession(),
				"SELECT o FROM JSlaveSynch o WHERE o.id.eid = ? AND o.autoSynched = ?", "*", true).iterate();
		String id = slave.getId();
		while (iterator.hasNext()) {
			JSlaveSynch slaveSynch = iterator.next();
			addSlaveSynch(id, slaveSynch.getId().getMid(), slaveSynch.getUri(), slaveSynch.getPostData());
		}
	}

	/**
	 * 服务区同步完成
	 * 
	 * @param slaveSynch
	 * @param updateTime
	 */
	@Transaction
	public void syncComplete(JSlaveSynch slaveSynch, long updateTime) {
		QueryDaoUtils.createQueryArray(BeanDao.getSession(),
				"UPDATE JSlaveSynch o SET o.synched = ? WHERE o.id.eid = ? AND o.id.mid= ? AND o.updateTime = ?", true,
				slaveSynch.getId().getEid(), slaveSynch.getId().getMid(), updateTime).executeUpdate();
	}

	/**
	 * 检查数据同步
	 */
	@Async(notifier = true)
	@Transaction(readOnly = true)
	@Schedule(fixedDelay = 300000, initialDelay = 20000)
	public void checkSyncs() {
		try {
			Thread.sleep(3000);

		} catch (Exception e) {
		}

		Iterator<JSlaveSynch> iterator = QueryDaoUtils.createQueryArray(BeanDao.getSession(),
				"SELECT o FROM JSlaveSynch o WHERE o.synched = ?", false).iterate();
		final UtilAtom atom = new UtilAtom();
		while (iterator.hasNext()) {
			final JSlaveSynch slaveSynch = iterator.next();
			SocketChannelContext context = MasterService.ME.getMasterReceiverContext().getServerContext().getChannelContexts()
					.get(slaveSynch.getId().getEid());
			if (context != null) {
				try {
					atom.increment();
					final long updateTime = slaveSynch.getUpdateTime();
					MasterService.ME.getMasterReceiverContext().sendDataBytes(context.getSocketChannel(),
							slaveSynch.getUri().getBytes(ContextUtils.getCharset()), true, false, slaveSynch.getPostData(), 60000,
							new CallbackMessagePack<String>() {

								@Override
								public void doWithBean(String bean, boolean ok, byte[] buffer, SocketAdapter adapter) {
									try {
										if (ok) {
											ME.syncComplete(slaveSynch, updateTime);
										}

									} finally {
										atom.decrement();
									}
								}

							});

				} catch (Exception e) {
					atom.decrement();
					LOGGER.error("checkSyncs " + context + " " + slaveSynch.getId() + " " + slaveSynch.getUri() + " => "
							+ slaveSynch.getPostData(), e);
				}
			}
		}

		atom.await();
	}

	@Transaction
	@DataQuery("SELECT o FROM JSlaveServer o WHERE o.synched = TRUE ORDER BY o.id DESC")
	public abstract List<JSlaveServer> getServers();

	/**
	 * @param channel
	 * @return
	 */
	@Transaction
	@DataQuery("SELECT o FROM JSlaveServerChannel o WHERE o.channel = ? AND o.server.synched = TRUE ORDER BY o.id DESC")
	public abstract List<JSlaveServerChannel> getServerChannelList(String channel);

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.absir.orm.hibernate.boost.IEntityMerge#merge(java.lang.String,
	 * java.lang.Object, com.absir.orm.hibernate.boost.IEntityMerge.MergeType,
	 * java.lang.Object)
	 */
	@Override
	public void merge(String entityName, JSlaveServer entity, com.absir.orm.hibernate.boost.IEntityMerge.MergeType mergeType,
			Object mergeEvent) {
		if (mergeType != MergeType.DELETE) {
			syncChange(entity);
			ME.syncServers();
		}
	}

	@Transaction
	@DataQuery("SELECT o FROM JSlaveServer o WHERE o.synched = FALSE")
	public abstract List<JSlaveServer> getNoSyncServers();

	/**
	 * 服务区同步变化
	 * 
	 * @param slaveServer
	 */
	@Transaction
	public void syncChange(JSlaveServer slaveServer) {
		QueryDaoUtils.createQueryArray(BeanDao.getSession(),
				"UPDATE JSlaveServer o SET o.synched = ?, o.updateTime = ? WHERE o.id = ?", false, System.currentTimeMillis(),
				slaveServer.getId()).executeUpdate();
	}

	/**
	 * 服务区同步完成
	 * 
	 * @param slaveServer
	 * @param updateTime
	 */
	@Transaction
	public void syncComplete(JSlaveServer slaveServer, long updateTime) {
		QueryDaoUtils.createQueryArray(BeanDao.getSession(),
				"UPDATE JSlaveServer o SET o.synched = ? WHERE o.id = ? AND o.updateTime = ?", true, slaveServer.getId(),
				updateTime).executeUpdate();
	}

	/**
	 * 同步服务区数据
	 * 
	 * @throws InterruptedException
	 */
	@Async(notifier = true)
	@Transaction(readOnly = true)
	@Schedule(fixedDelay = 300000, initialDelay = 20000)
	public void syncServers() {
		try {
			Thread.sleep(3000);

		} catch (Exception e) {
		}

		List<JSlaveServer> slaveServers = ME.getNoSyncServers();
		final UtilAtom atom = new UtilAtom();
		for (final JSlaveServer slaveServer : slaveServers) {
			JSlave host = slaveServer.getHost();
			if (host != null) {
				SocketChannelContext context = MasterService.ME.getMasterReceiverContext().getServerContext().getChannelContexts()
						.get(host.getId());
				if (context != null) {
					try {
						atom.increment();
						final long updateTime = slaveServer.getUpdateTime();
						MasterService.ME.getMasterReceiverContext().sendData(context.getSocketChannel(), "api/slave/sync",
								slaveServer, new CallbackMessagePack<String>() {

									@Override
									public void doWithBean(String bean, boolean ok, byte[] buffer, SocketAdapter adapter) {
										try {
											if (ok) {
												syncComplete(slaveServer, updateTime);
											}

										} finally {
											atom.decrement();
										}
									}

								});

					} catch (Exception e) {
						atom.decrement();
						LOGGER.error("syncServers " + context + " " + host.getId() + " => " + slaveServer, e);
					}
				}
			}
		}

		atom.await();
	}
}
